package com.lmf.rabbitmq.eight;

import com.lmf.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 死信的三大来源： 1.消息被拒绝     2.消息过期       3.正常队列长度达到最长，多出的消息会被转发到死信队列
 *
 * 模拟实现死信队列：（消费者01）
 *          consumer 从 queue 取出消息 进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，
 *          这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。死信队列可以再通过其他消费者接收
 *
 */
public class Consumer01 {

    //定义普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //定义死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";
    //定义普通队列名称
    private static final String NORMAL_NAME = "normal_name";
    //定义死信队列名称
    private static final String DEAD_NAME = "dead_name";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();
        //声明普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        /**
         *   如何将正常队列里面的信息转发给死信队列?
         *   需要在正常队列里面，声明arguments参数:
         *   x-dead-letter-exchange ：死信交换机名称
         *   x-dead-letter-routing-key  ： 与死信交换机绑定的routing-key参数
         *   x-max-length   ： 设置当前队列的最大长度
         *   x-message-ttl ：设置队列里面，每条消息过期时间
         */
        //正常队列绑定死信队列信息
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        arguments.put("x-dead-letter-routing-key","lisi");
//        arguments.put("x-max-length",6);

        //声明普通队列
        channel.queueDeclare(NORMAL_NAME,false,false,false,arguments);
        //声明死信队列：不用声明argument参数
        channel.queueDeclare(DEAD_NAME,false,false,false,null);

        //将普通队列与普通交换机绑定
        channel.queueBind(NORMAL_NAME,NORMAL_EXCHANGE,"zhangsan");
        //将死信队列与死信交换机绑定
        channel.queueBind(DEAD_NAME,DEAD_EXCHANGE,"lisi");

        //消息接收成功，回调函数
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //打印
                String str = new String(message.getBody());
                if (str.equals("5")) {
                    System.out.println("普通队列接收成功的消息：标识为：" + consumerTag + "； 信息内容为：" + str +
                            "该消息以已经被拒绝，会被转发至死信队列");
                    //拒绝消息
                    /**
                     * basicReject():拒绝某个消息
                     *      1.被拒绝消息的信道标识
                     *      2.是否重新投入正常队列，一般都false，直接转发至死信队列
                     */
                    channel.basicReject(message.getEnvelope().getDeliveryTag(),false);

                }
                System.out.println("普通队列接收成功的消息：标识为：" + consumerTag + "； 信息内容为：" + str);
                /**
                 * basicAck():手动应答消息
                 *      1.消息标记 tag
                 *      2.是否批量应答未应答消息,会将整个队列的所有消息全部自动应答
                 */
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            }
        };
        //接收普通队列消息
        System.out.println("c1普通队列等待接收消息........... ");
        //演示拒绝某个消息：必须开启手动应答
        channel.basicConsume(NORMAL_NAME,false,deliverCallback,consumerTag -> {});
    }

}
